package com.zf.nospring.rabbitmq.nofair;

import com.rabbitmq.client.*;
import com.zf.nospring.rabbitmq.utils.RabbitUtils;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 多个消费者订阅同一个队列，如何实现不公平分发（公平消费）
 * <p>
 * 对于多个消费者消费同一个队列的情况，rabbitmq默认采用轮询策略，把消息均匀的分发给每个消费者，也就是公平分发（公平消费）
 * <p>
 * 可以使用channel.basicQos(prefetchCount) 可以改变消息分发的策略，实现不公平分发（不公平消费），即能者多劳。
 * <p>
 * 要使 channel.basicQos(prefetchCount) 生效必须要关闭自动确认，改为手动确认
 */
public class ConsumerNoFair {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = RabbitUtils.getDefaultConnection();
        Channel channel1 = connection.createChannel();
        Channel channel2 = connection.createChannel();
        Channel channel3 = connection.createChannel();

        /* channel.basicQos(prefetchCount) 设置通道中所能容纳的未确认消息的最大数量，同时也是设置rabbitmq服务器能一次向该通道中分发的消息的最大数量，
            该设置只在手动ack的情况下生效,自动ack不生效。
          1、只有在通道中的未确认消息数量小于最大数量时，rabbitmq才可以继续向该通道中分发消息。
          2、一旦通道中的未确认消息数量等于最大数量时，rabbitmq就不能再向该通道中分发消息了，
              只能等待该通道的消息被确认，这样未确认消息的数量就减少了，这样就再次回到步骤 1，
              或者如果还有其他通道消费该队列就可以把剩余的消息分发给其他通道；
         */
        // channel1 不设置，默认是0 表示通道没有最大未确认消息数量的限制，这样 channel1可以拿到最多数量的消息
        channel1.basicConsume("test_no_fair", false, getDefaultConsumer(channel1));

        // channel2设置为1，表示该通道中所能容纳的未确认消息数量的最大值是 1，那么只有通道中的消息被确认消费后，rabbitmq才会再次向该通代分发消息
        // channel2中的拿到的消息数量最少
        channel2.basicQos(1);
        channel2.basicConsume("test_no_fair", false, getDefaultConsumer(channel2));

        // channel3设置为3，表示通道中所能容纳的未确认消息数量的最大值是 3
        // channel3中拿到的消息数量居中
        channel3.basicQos(3);
        channel3.basicConsume("test_no_fair", false, getDefaultConsumer(channel3));

    }

    /**
     * 创建消费者的消费逻辑
     *
     * @param channel 通道
     */
    public static Consumer getDefaultConsumer(Channel channel) {
        // 创建消费者的消费逻辑
        return new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                long deliveryTag = envelope.getDeliveryTag();
                // 获取当前通道
                Channel ch = getChannel();
                System.out.printf("channel: %s\n consumerTag：%s，deliveryTag: %d，msg: %s\n", ch.toString(), consumerTag, deliveryTag, new String(body));
                // 手动向 rabbitmq 确认消息已被消费
                ch.basicAck(deliveryTag, false);
            }
        };
    }
}
